package com.stars.easyms.mq.core.activemq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.stars.easyms.mq.asyn.AsynReceiveMQMessage;
import com.stars.easyms.mq.constant.MQConstants;
import com.stars.easyms.mq.core.MQListener;
import com.stars.easyms.mq.dto.MQInfoDTO;
import com.stars.easyms.mq.dto.MQMessageDTO;
import com.stars.easyms.mq.enums.DestinationType;
import com.stars.easyms.mq.enums.MQType;
import com.stars.easyms.mq.properties.EasyMsMQProperties;
import com.stars.easyms.mq.service.MQMessageService;
import com.stars.easyms.base.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

import javax.jms.*;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.sql.Timestamp;
import java.util.*;

/**
 * ActiveMQ监听类
 *
 * @author guoguifang
 * @date 2018-04-23 13:54
 * @since 1.0.0
 */
public class ActiveMQListener implements MQListener {

    private static Logger logger = LoggerFactory.getLogger(ActiveMQListener.class);
    private MQInfoDTO mqInfoDTO;
    private String key;
    private ConnectionFactory connectionFactory;
    private MQMessageService mqMessageService;
    private AsynReceiveMQMessage asynReceiveMQMessage;
    private String applicationId;
    private Connection connection;
    private volatile boolean running;

    @Override
    public synchronized void start() {
        if (this.running) {
            logger.error("The MQ listener({}) is already running!", key);
            return;
        }

        Session session = null;
        try {
            this.connection = connectionFactory.createConnection();
            this.connection.start();
            session = this.connection.createSession(false, 2);
            Destination destination;
            if (DestinationType.QUEUE == mqInfoDTO.getDestinationType()) {
                destination = session.createQueue(key);
            } else {
                destination = session.createTopic(key);
            }
            MessageConsumer consumer = session.createConsumer(destination);
            consumer.setMessageListener(message -> {
                List<MQMessageDTO> mqMessageDTOList = null;
                Object messageObj = null;
                try {
                    if (message instanceof TextMessage) {
                        // 如果是String消息
                        TextMessage tm = (TextMessage) message;
                        String text = tm.getText();
                        // 首先转换成MQMessageDTO，如果转换后messageId不为空则表示是内部mq调用
                        try {
                            mqMessageDTOList = JSON.parseObject(text, new TypeReference<List<MQMessageDTO>>() {
                            });
                            messageObj = mqMessageDTOList;
                        } catch (Exception e) {
                            logger.debug("消息转换List<MQMessageDTO>失败!", e);
                        }
                        if (mqMessageDTOList == null) {
                            Class<?> parameterClass = ActiveMQListener.this.mqInfoDTO.getParameterClass();
                            if (String.class.isAssignableFrom(parameterClass)) {
                                messageObj = text;
                            } else {
                                Type parameterType = ActiveMQListener.this.mqInfoDTO.getParameterType();
                                if (!(parameterType instanceof ParameterizedType)) {
                                    messageObj = JSON.parseObject(text, parameterClass);
                                } else {
                                    messageObj = GenericTypeUtil.parseObject(text, (ParameterizedType) parameterType, parameterClass);
                                }
                            }
                        }
                    } else if (message instanceof MapMessage) {
                        // 如果是Map消息
                        MapMessage mm = (MapMessage) message;
                        Map<String, Object> map = new HashMap<>(64);
                        Enumeration enums = mm.getMapNames();
                        while (enums.hasMoreElements()) {
                            String element = (String) enums.nextElement();
                            map.put(element, mm.getObject(element));
                        }
                        messageObj = map;
                    } else if (message instanceof ObjectMessage) {
                        // 如果是Object消息
                        ObjectMessage om = (ObjectMessage) message;
                        messageObj = om.getObject();
                    } else if (message instanceof BytesMessage || message instanceof StreamMessage) {
                        List<Byte> byteList = new ArrayList<>();
                        byte[] byteArray = new byte[1024];
                        if (message instanceof BytesMessage) {
                            // 如果是bytes消息
                            BytesMessage bm = (BytesMessage) message;
                            while (bm.readBytes(byteArray) != -1) {
                                for (byte b : byteArray) {
                                    byteList.add(b);
                                }
                            }
                        } else {
                            // 如果是Stream消息
                            StreamMessage sm = (StreamMessage) message;
                            while (sm.readBytes(byteArray) != -1) {
                                for (byte b : byteArray) {
                                    byteList.add(b);
                                }
                            }
                        }
                        byte[] bytes = new byte[byteList.size()];
                        int index = 0;
                        for (Byte b : byteList) {
                            bytes[index++] = b;
                        }
                        messageObj = SerializeUtil.unserialize(bytes);
                    }

                    String messageStr = JsonUtil.toJSONString(messageObj);

                    // 记录待接收处理消息日志
                    logger.info("接收到目标({})待处理MQ消息,消息内容:{}", key, messageStr);

                    if (mqMessageDTOList != null) {
                        for (MQMessageDTO mqMessageDTO : mqMessageDTOList) {
                            mqMessageDTO.setId(mqMessageService.getId());
                            mqMessageDTO.setType(MQType.RECEIVE.getCode());
                            mqMessageDTO.setDestinationType(mqInfoDTO.getDestinationType().getCode());
                            mqMessageDTO.setDestSys(applicationId);
                            mqMessageDTO.setStatus(MQConstants.MQ_STATUS_PROCESSING);
                            mqMessageDTO.setFailCount(0L);
                            mqMessageDTO.setRedeliveryPolicy(mqInfoDTO.getRedeliveryPolicy().getCode());
                            mqMessageDTO.setRedeliveryDelay(mqInfoDTO.getRedeliveryDelay());
                            mqMessageDTO.setFireTime(new Timestamp(System.currentTimeMillis()));
                            asynReceiveMQMessage.add(mqMessageDTO);
                        }
                    } else {
                        MQMessageDTO mqMessageDTO = new MQMessageDTO();
                        mqMessageDTO.setId(mqMessageService.getId());
                        mqMessageDTO.setMessageId(mqMessageService.getMessageId());
                        mqMessageDTO.setKey(key);
                        mqMessageDTO.setMessage(messageStr);
                        mqMessageDTO.setType(MQType.RECEIVE.getCode());
                        mqMessageDTO.setDestinationType(mqInfoDTO.getDestinationType().getCode());
                        mqMessageDTO.setDestSys(applicationId);
                        mqMessageDTO.setStatus(MQConstants.MQ_STATUS_PROCESSING);
                        mqMessageDTO.setFailCount(0L);
                        mqMessageDTO.setRedeliveryPolicy(mqInfoDTO.getRedeliveryPolicy().getCode());
                        mqMessageDTO.setRedeliveryDelay(mqInfoDTO.getRedeliveryDelay());
                        mqMessageDTO.setFireTime(new Timestamp(System.currentTimeMillis()));
                        asynReceiveMQMessage.add(mqMessageDTO);
                    }
                    message.acknowledge();
                } catch (Exception e) {
                    logger.error("接收目标({})MQ消息失败,消息内容:{}!", key, JSON.toJSONString(message), e);
                }
            });
            this.running = true;
            logger.info("The MQ listener({}) destination type({}) start up success!", key, mqInfoDTO.getDestinationType());
        } catch (Exception e) {
            logger.error("The MQ listener({}) destination type({}) start up failure!", key, mqInfoDTO.getDestinationType(), e);
            if (session != null) {
                try {
                    session.close();
                } catch (Exception e1) {
                    logger.error("The MQ listener({}) session close failure!", key, e1);
                }
            }
            if (this.connection != null) {
                try {
                    this.connection.close();
                } catch (Exception e1) {
                    logger.error("The MQ listener({}) connection close failure!", key, e1);
                }
            }
        }
    }

    @Override
    public synchronized void stop() {
        if (!this.running) {
            logger.error("The MQ listener({}) is not started!", key);
            return;
        }
        try {
            this.connection.stop();
            this.running = false;
            logger.info("The MQ listener({}) shut down success!", key);
        } catch (Exception e) {
            logger.error("The MQ listener({}) shut down failure!", key, e);
        }
    }

    @Override
    public synchronized boolean isRunning() {
        return this.running;
    }

    public ActiveMQListener(ApplicationContext applicationContext, String applicationId, EasyMsMQProperties easyMsMQProperties, MQMessageService mqMessageService, AsynReceiveMQMessage asynReceiveMQMessage, MQInfoDTO mqInfoDTO, String key) {
        this.connectionFactory = BasicActiveMQConnectionFactory.getActiveMQConnectionFactory(applicationContext, easyMsMQProperties);
        this.mqMessageService = mqMessageService;
        this.asynReceiveMQMessage = asynReceiveMQMessage;
        this.applicationId = applicationId;
        this.mqInfoDTO = mqInfoDTO;
        this.key = key;
    }
}
